package beautiful.butterfly.drds.data_exchange.util;


import beautiful.butterfly.drds.data_exchange.error_code.DBUtilErrorCode;
import beautiful.butterfly.drds.data_exchange.error_code.DataExchangeException;
import beautiful.butterfly.drds.data_exchange.read_and_write_data.reader.Key;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.ImmutableTriple;
import org.apache.commons.lang3.tuple.Triple;

import java.sql.*;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

@Slf4j
public final class Jdbcs1
{


    private static final ThreadLocal<ExecutorService> THREAD_LOCAL = new ThreadLocal<ExecutorService>()
    {
        @Override
        protected ExecutorService initialValue()
        {
            return Executors.newFixedThreadPool(1, new ThreadFactoryBuilder()
                    .setNameFormat("THREAD_LOCAL-%d")
                    .setDaemon(true)
                    .build());
        }
    };

    private Jdbcs1()
    {
    }

    public static String chooseJdbcUrlWithRetry(
            final List<String> jdbcUrlList, final String username,
            final String password, final List<String> preSqlList,
            final boolean checkSlave)
    {

        if (null == jdbcUrlList || jdbcUrlList.isEmpty())
        {
            throw DataExchangeException.asDataExchangeException(
                    DBUtilErrorCode.CONF_ERROR,
                    String.format("您的jdbcUrl的配置信息有错, 因为jdbcUrl[%s]不能为空. 请检查您的配置并作出修改.",
                            StringUtils.join(jdbcUrlList, ",")));
        }

        try
        {
            return RetryableTasks.executeWithRetry(new Callable<String>()
            {

                public String call() throws Exception
                {
                    boolean ok = false;
                    for (String jdbcUrl : jdbcUrlList)
                    {
                        if (StringUtils.isNotBlank(jdbcUrl))
                        {
                            jdbcUrl = jdbcUrl.trim();
                            if (null != preSqlList && !preSqlList.isEmpty())
                            {
                                ok = getConnection(
                                        jdbcUrl, username, password, preSqlList);
                            } else
                            {
                                ok = getConnection(
                                        jdbcUrl, username, password, checkSlave);
                            }
                            if (ok)
                            {
                                return jdbcUrl;
                            }
                        }
                    }
                    throw new Exception("DataX无法连接对应的数据库，可能原因是：1) 配置的ip/port/database/jdbc错误，无法连接。2) 配置的username/password错误，鉴权失败。请和DBA确认该数据库的连接信息是否正确。");
//                    throw new Exception(DBUtilErrorCode.JDBC_NULL.toString());
                }
            }, 7, 1000L, true);
            //warn: 7 means 2 minutes
        } catch (Exception e)
        {
            throw DataExchangeException.asDataExchangeException(
                    DBUtilErrorCode.CONN_DB_ERROR,
                    String.format("数据库连接失败. 因为根据您配置的连接信息,无法从:%s 中找到可连接的jdbcUrl. 请检查您的配置并作出修改.",
                            StringUtils.join(jdbcUrlList, ",")), e);
        }
    }

    public static String chooseJdbcUrl(
            final List<String> jdbcUrlList, final String username,
            final String password, final List<String> preSqlList,
            final boolean checkSlave) throws DataExchangeException
    {

        if (null == jdbcUrlList || jdbcUrlList.isEmpty())
        {
            throw DataExchangeException.asDataExchangeException(
                    DBUtilErrorCode.CONF_ERROR,
                    String.format("您的jdbcUrl的配置信息有错, 因为jdbcUrl[%s]不能为空. 请检查您的配置并作出修改.",
                            StringUtils.join(jdbcUrlList, ",")));
        }

        boolean ok = false;
        for (String jdbcUrl : jdbcUrlList)
        {
            if (StringUtils.isNotBlank(jdbcUrl))
            {
                jdbcUrl = jdbcUrl.trim();
                if (null != preSqlList && !preSqlList.isEmpty())
                {
                    ok = getConnection(jdbcUrl, username, password, preSqlList);
                } else
                {
                    try
                    {
                        ok = getConnection(jdbcUrl, username, password, checkSlave);
                    } catch (Exception e)
                    {
                        throw DataExchangeException.asDataExchangeException(
                                DBUtilErrorCode.CONN_DB_ERROR,
                                String.format("数据库连接失败. 因为根据您配置的连接信息,无法从:%s 中找到可连接的jdbcUrl. 请检查您的配置并作出修改.",
                                        StringUtils.join(jdbcUrlList, ",")), e);
                    }
                }
                if (ok)
                {
                    return jdbcUrl;
                }
            }
        }
        throw DataExchangeException.asDataExchangeException(
                DBUtilErrorCode.CONN_DB_ERROR,
                String.format("数据库连接失败. 因为根据您配置的连接信息,无法从:%s 中找到可连接的jdbcUrl. 请检查您的配置并作出修改.",
                        StringUtils.join(jdbcUrlList, ",")));
    }

    /**
     * 检查slave的库中的数据是否已到凌晨00:00
     * 如果slave同步的数据还未到00:00返回false
     * 否则范围true
     *
     * @author ZiChi
     * @version 1.0 2014-12-01
     */
    private static boolean isSlaveBehind(Connection connection)
    {
        try
        {
            ResultSet resultSet = Jdbc.executeQuery(connection, "show variables like 'read_only'");
            if (Jdbcs1.asyncResultSetNext(resultSet))
            {
                String $Value = resultSet.getString("Value");
                if ("ON".equalsIgnoreCase($Value))
                { //备库
                    ResultSet resultSet1 = Jdbc.executeQuery(connection, "show slave status");
                    if (Jdbcs1.asyncResultSetNext(resultSet1))
                    {
                        String $Slave_IO_Running = resultSet1.getString("slave_io_running");
                        String $Slave_SQL_Running = resultSet1.getString("slave_sql_running");
                        long $Seconds_Behind_Master = resultSet1.getLong("seconds_behind_master");
                        if ("Yes".equalsIgnoreCase($Slave_IO_Running) && "Yes".equalsIgnoreCase($Slave_SQL_Running))
                        {
                            ResultSet resultSet2 = Jdbc.executeQuery(connection, "select timestampdiff(second, curdate(), now())");
                            Jdbcs1.asyncResultSetNext(resultSet2);
                            long secondsOfDay = resultSet2.getLong(1);
                            return $Seconds_Behind_Master > secondsOfDay;
                        } else
                        {
                            return true;
                        }
                    } else
                    {
                        log.warn("SHOW SLAVE STATUS has no result");
                    }
                }
            } else
            {
                log.warn("SHOW VARIABLES like 'read_only' has no result");
            }
        } catch (Exception e)
        {
            log.warn("checkSlave failed, errorMessage:[{}].", e.getMessage());
        }
        return false;
    }


    //
    public static List<String> getColumnNameList(
            String jdbcUrl, String username, String password, String tableName) throws SQLException
    {
        Connection connection = Jdbc.getConnectionWithRetry(jdbcUrl, username, password);
        return getColumnNameList(connection, tableName);
    }

    public static List<String> getColumnNameList(Connection connection, String tableName)
    {
        List<String> columnNameList = new ArrayList<String>();
        Statement statement = null;
        ResultSet resultSet = null;
        String queryColumnSql = null;
        try
        {
            statement = connection.createStatement();
            queryColumnSql = String.format("select * from %s where 1=2", tableName);
            resultSet = statement.executeQuery(queryColumnSql);
            ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
            for (int i = 0, columnCount = resultSetMetaData.getColumnCount(); i < columnCount; i++)
            {
                columnNameList.add(resultSetMetaData.getColumnName(i + 1));
            }

        } catch (SQLException e)
        {
            throw RdbmsException.asQueryException(e, queryColumnSql, tableName, null);
        } finally
        {
            Jdbc.close(resultSet, statement, connection);
        }

        return columnNameList;
    }


    /**
     * @return Left:ColumnName Middle:ColumnType Right:ColumnTypeName
     */
    public static Triple<List<String>, List<Integer>, List<String>> getColumnMetaDataList(Connection connection, String tableName, String columnName)
    {
        Statement statement = null;
        ResultSet resultSet = null;

        Triple<List<String>, List<Integer>, List<String>> triple = new ImmutableTriple<List<String>, List<Integer>, List<String>>(//
                new ArrayList<String>(),//
                new ArrayList<Integer>(),//
                new ArrayList<String>());//
        try
        {
            statement = connection.createStatement();
            String sql = "select " + columnName + " from " + tableName + " where 1=2";
            resultSet = statement.executeQuery(sql);
            //
            ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
            for (int i = 0, columnCount = resultSetMetaData.getColumnCount(); i < columnCount; i++)
            {

                triple.getLeft().add(resultSetMetaData.getColumnName(i + 1));
                triple.getMiddle().add(resultSetMetaData.getColumnType(i + 1));
                triple.getRight().add(resultSetMetaData.getColumnTypeName(i + 1));
            }
            return triple;

        } catch (SQLException e)
        {
            throw DataExchangeException
                    .asDataExchangeException(DBUtilErrorCode.GET_COLUMN_INFO_FAILED,
                            String.format("获取表:%s 的字段的元信息时失败. 请联系 DBA 核查该库、表信息.", tableName), e);
        } finally
        {
            Jdbc.close(resultSet, statement, null);
        }
    }

    public static boolean getConnection(String url, String username, String password, boolean checkSlave)
    {
        Connection connection = null;

        try
        {
            connection = Jdbc.getConnection(url, username, password);
            if (connection != null)
            {
                if (checkSlave)
                {
                    boolean ok = !isSlaveBehind(connection);
                    return ok;
                } else
                {
                    return true;
                }
            }
        } catch (Exception e)
        {
            log.warn("test connection of [{}] failed, for {}.", url,
                    e.getMessage());
        } finally
        {
            Jdbc.close(null, connection);
        }
        return false;
    }

    public static boolean getConnection(String url, String username, String password, List<String> preCheckSqlList)
    {
        Connection connection = null;
        try
        {
            connection = Jdbc.getConnection(url, username, password);
            if (null != connection)
            {
                for (String preCheckSql : preCheckSqlList)
                {
                    if (doPreCheck(connection, preCheckSql) == false)
                    {
                        log.warn("doPreCheck failed.");
                        return false;
                    }
                }
                return true;
            }
        } catch (Exception e)
        {
            log.warn("test connection of [{}] failed, for {}.", url,
                    e.getMessage());
        } finally
        {
            Jdbc.close(null, connection);
        }

        return false;
    }


    private static boolean doPreCheck(Connection connection, String preCheckSql)
    {
        ResultSet resultSet = null;
        try
        {
            resultSet = Jdbc.executeQuery(connection, preCheckSql);

            int checkResult = -1;
            if (Jdbcs1.asyncResultSetNext(resultSet))
            {
                checkResult = resultSet.getInt(1);
                if (Jdbcs1.asyncResultSetNext(resultSet))
                {
                    log.warn(
                            "pre check failed. It should return one result:0, pre:[{}].",
                            preCheckSql);
                    return false;
                }

            }

            if (0 == checkResult)
            {
                return true;
            }

            log.warn(
                    "pre check failed. It should return one result:0, pre:[{}].",
                    preCheckSql);
        } catch (Exception e)
        {
            log.warn("pre check failed. pre:[{}], errorMessage:[{}].", preCheckSql,
                    e.getMessage());
        } finally
        {
            Jdbc.close(resultSet);
        }
        return false;
    }

    // warn:until now, only oracle need to handle session config.
    public static void dealWithSessionConfig(Connection connection,
                                             Configuration configuration, String message) throws SQLException
    {
        List<String> sessionConfig = null;
        sessionConfig = configuration.getList(Key.session,
                new ArrayList<String>(), String.class);
        Jdbcs1.doDealWithSessionConfig(connection, sessionConfig, message);

    }

    private static void doDealWithSessionConfig(Connection connection,
                                                List<String> sessions, String message) throws SQLException
    {
        if (null == sessions || sessions.isEmpty())
        {
            return;
        }

        Statement statement;
        statement = connection.createStatement();

        for (String sessionSql : sessions)
        {
            log.info("execute sql:[{}]", sessionSql);
            try
            {
                Jdbc.execute(statement, sessionSql);
            } catch (SQLException e)
            {
                throw DataExchangeException.asDataExchangeException(
                        DBUtilErrorCode.SET_SESSION_ERROR, String.format(
                                "session配置有误. 因为根据您的配置执行 session 设置失败. 上下文信息是:[%s]. 请检查您的配置并作出修改.", message), e);
            }
        }
        Jdbc.close(statement, null);
    }


    /**
     * 异步获取resultSet的next(),注意，千万不能应用在数据的读取中。只能用在meta的获取
     */
    public static boolean asyncResultSetNext(final ResultSet resultSet)
    {
        return asyncResultSetNext(resultSet, 3600);
    }

    public static boolean asyncResultSetNext(final ResultSet resultSet, int timeout)
    {
        Future<Boolean> future = THREAD_LOCAL.get().submit(new Callable<Boolean>()
        {
            public Boolean call() throws Exception
            {
                return resultSet.next();
            }
        });
        try
        {
            return future.get(timeout, TimeUnit.SECONDS);
        } catch (Exception e)
        {
            throw DataExchangeException.asDataExchangeException(
                    DBUtilErrorCode.RS_ASYNC_ERROR, "异步获取ResultSet失败", e);
        }
    }


}
